package cn.itcast.map;

import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import cn.itcast.util.DbUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Map;

/**
 * @Date 2021
 * 个股K线数据转换（日，周，月）
 */
public class StockKlineMap extends RichMapFunction<StockBean, Row> {
    /**
     * 一、初始化
     * 1.创建构造方法
     * 入参：kType：K线类型
     * firstTxdate：周期首个交易日
     * 2.获取交易日历表交易日数据
     * 3.获取周期首个交易日和T日
     * 4.获取K(周、月)线下的汇总表数据（高、低、成交量、金额）
     */
    // 1.创建构造方法
    String kType;//日K：1，周K：2，月K：3
    String firstTxdate;

    Map<String, Map<String, Object>> klineMap = null;
    String tradeDate = null;
    String firstTradeDate = null;

    public StockKlineMap(String kType, String firstTxdate) {
        this.kType = kType;
        this.firstTxdate = firstTxdate;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //查询数据库查询日期 日周年表工具类 得到本周第一天
        String sql = "select * from tcc_date where  trade_date =CURDATE()";
        Map<String, String> stringStringMap = DbUtil.queryKv(sql);
        //3.获取周期首个交易日和T日
        //如果查询的是日K数据，那么T日和周期内的首个交易日，日期一样
        //如果是月K和周K，那么T日的时间 > 周期内的首个交易日间
        firstTradeDate = stringStringMap.get(firstTxdate); //周期首个交易日
        tradeDate = stringStringMap.get("trade_date");  //T日
        //4.获取K(周、月)线下的汇总表数据（高、低、成交量、金额）
        String sql2 = "SELECT\n" +
                "  sec_code,\n" +
                "  MAX(high_price) AS high_price,\n" +
                "  MIN(low_Price) AS low_Price,\n" +
                "  SUM(trade_vol) AS trade_vol,\n" +
                "  SUM(trade_amt) AS trade_amt\n" +
                "FROM bdp_quot_stock_kline_day\n" +
                "WHERE trade_date \n" +
                "      BETWEEN " + firstTradeDate + " AND " + tradeDate + " \n" +
                "GROUP BY 1";
        klineMap = DbUtil.query("sec_code", sql2);
    }

    /**
     * 二、业务处理
     * 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
     * 2.获取T日和周首次交易日时间,转换成long型
     * 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
     * 4.获取周/月K数据：成交量、成交额、高、低
     * 5.高、低价格比较
     * 6.计算成交量、成交额
     * 7.计算均价
     * 8.封装数据Row
     */
    @Override
    public Row map(StockBean value) throws Exception {
//        1.获取个股部分数据（前收、收、开盘、高、低、量、金额）

        BigDecimal preClosePrice = value.getPreClosePrice();
        BigDecimal openPrice = value.getOpenPrice();
        BigDecimal closePrice = value.getClosePrice();
        BigDecimal highPrice = value.getHighPrice();
        BigDecimal lowPrice = value.getLowPrice();
        Long tradeAmtDay = value.getTradeAmtDay();
        Long tradeVolDay = value.getTradeVolDay();
        //2获取T日和周首次交易日时间,转换成long型
        Long tradeTime = DateUtil.stringToLong(tradeDate, Constant.format_yyyymmdd);
        Long firstTradeTime = DateUtil.stringToLong(firstTradeDate, Constant.format_yyyymmdd);
//        比较周期首个交易日和当天交易日大小，判断是否是周、月K线
        if (firstTradeTime < tradeTime && kType.equals("2") || kType.equals("3")) {
        //一定是周 月
//             * 4.获取周/月K数据：成交量、成交额、高、低
            Map<String, Object> stringObjectMap = klineMap.get(value.getSecCode());
            /*MAX(high_price) AS high_price,\n" +
                "  MIN(low_Price) AS low_Price,\n" +
                "  SUM(trade_vol) AS trade_vol,\n" +
                "  SUM(trade_amt) AS trade_amt\n" +*/
            if (stringObjectMap!=null) {
                //最高价
                BigDecimal highPriceLast = new BigDecimal(stringObjectMap.get("high_price").toString());
                //最低价
                BigDecimal lowPriceLast = new BigDecimal(stringObjectMap.get("low_Price").toString());
                //成交量
                Long tradeVolLast = Long.valueOf(stringObjectMap.get("trade_vol").toString());
//            成交金额
                Long tradeAmtLast = Long.valueOf(stringObjectMap.get("trade_amt").toString());
                //.高、低价格比较
                if (highPrice.compareTo(highPriceLast) == -1) {
                    highPrice = highPriceLast;
                }
                if (lowPrice.compareTo(lowPriceLast) == 1) {
                    lowPrice = lowPriceLast;
                }
                //6.计算成交额
                tradeAmtDay+=tradeAmtLast;
                //成交量、
                tradeVolDay+=tradeVolLast;
            }
        }
        //计算均价
        BigDecimal divide = new BigDecimal(tradeAmtDay).divide(new BigDecimal(tradeVolDay), 2, RoundingMode.HALF_UP);
        // 8.封装数据Row
        Row row = new Row(13);
        row.setField(0,new Timestamp(value.getEventTime()));
        row.setField(1,tradeDate);
        row.setField(2,value.getSecCode());
        row.setField(3,value.getSecName());
        row.setField(4,kType);
        row.setField(5,preClosePrice);
        row.setField(6,openPrice);
        row.setField(7,highPrice);
        row.setField(8,lowPrice);
        row.setField(9,closePrice);
        row.setField(10,divide);
        row.setField(11,tradeVolDay);
        row.setField(12,tradeAmtDay);
        return row;
    }
}
